package com.atguigu.util;

import com.alibaba.fastjson.JSONObject;
import com.atguigu.common.Constant;
import org.apache.hadoop.conf.Configuration;
import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.HBaseConfiguration;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;

import java.io.IOException;
import java.util.ArrayList;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.ExecutionException;

public class HBaseUtil {

    //获取连接
    public static Connection getConnection() throws IOException {
        Configuration configuration = HBaseConfiguration.create();
        return ConnectionFactory.createConnection(configuration);
    }

    //获取异步连接
    public static AsyncConnection getAsyncConnection() throws ExecutionException, InterruptedException {
        Configuration configuration = HBaseConfiguration.create();
        return ConnectionFactory.createAsyncConnection(configuration).get();
    }

    /**
     * @param connection HBase连接
     * @param nameSpace  命名空间
     * @param table      表名
     * @param splitKeys  分区键     "00|,01|,02|"
     * @param cfs        列族  可以是多个
     */
    public static void createTable(Connection connection, String nameSpace, String table, byte[][] splitKeys, String... cfs) throws IOException {

        //对列族进行校验
        int length = cfs.length;
        if (length <= 0) {
            throw new RuntimeException("请设置至少一个列族！");
        }

        //判断表是否存在
        Admin admin = connection.getAdmin();
        TableName tableName = TableName.valueOf(nameSpace + ":" + table);
        if (admin.tableExists(tableName)) {
            System.out.println("待创建表" + table + "已经存在！");
            return;
        }

        //建表操作
        ArrayList<ColumnFamilyDescriptor> columnFamilyDescriptors = new ArrayList<>();
        for (String cf : cfs) {
            ColumnFamilyDescriptor familyDescriptor = ColumnFamilyDescriptorBuilder.newBuilder(cf.getBytes())
                    .build();
            columnFamilyDescriptors.add(familyDescriptor);
        }
        TableDescriptor tableDescriptor = TableDescriptorBuilder.newBuilder(tableName)
                .setColumnFamilies(columnFamilyDescriptors)
                .build();
        if (splitKeys == null) {
            admin.createTable(tableDescriptor);
        } else {
            admin.createTable(tableDescriptor, splitKeys);
        }

        //释放资源
        admin.close();
    }

    //删表
    public static void dropTable(Connection connection, String nameSpace, String table) throws IOException {

        Admin admin = connection.getAdmin();

        //判断表是否存在
        TableName tableName = TableName.valueOf(nameSpace + ":" + table);
        if (!admin.tableExists(tableName)) {
            System.out.println("待删除表" + table + "不存在！");
            return;
        } else {
            //执行删除操作
            admin.disableTable(tableName);
            admin.deleteTable(tableName);
        }

        //释放资源
        admin.close();
    }

    //sinkExtend:"00|,01|,02|,03|..."
    public static byte[][] getSplitKeys(String sinkExtend) {
        if (sinkExtend == null) {
            return null;
        } else {
            String[] split = sinkExtend.split(",");
            byte[][] splitKeys = new byte[split.length][];
            for (int i = 0; i < split.length; i++) {
                splitKeys[i] = split[i].getBytes();
            }
            return splitKeys;
        }
    }

    //put 'bigdata:stu','1001','info:name','zhangsan'
    public static void putData(Connection connection, String nameSpace, String table, String rowKey, String cf, String cn, String value) throws IOException {
        //获取表对象
        TableName tableName = TableName.valueOf(nameSpace + ":" + table);
        Table connectionTable = connection.getTable(tableName);
        //创建Put对象
        Put put = new Put(rowKey.getBytes());
        put.addColumn(cf.getBytes(), cn.getBytes(), value.getBytes());
        //执行插入数据操作
        connectionTable.put(put);
        //释放资源
        connectionTable.close();
    }

    public static void putJsonData(Connection connection, String nameSpace, String table, String rowKey, String cf, JSONObject data) throws IOException {
        //获取表对象
        TableName tableName = TableName.valueOf(nameSpace + ":" + table);
        Table connectionTable = connection.getTable(tableName);
        //创建Put对象
        Put put = new Put(rowKey.getBytes());

        //遍历数据集,将其放入Put对象
        Set<Map.Entry<String, Object>> entries = data.entrySet();
        for (Map.Entry<String, Object> entry : entries) {
            Object value = entry.getValue();
            if (value != null) {
                put.addColumn(cf.getBytes(), entry.getKey().getBytes(), value.toString().getBytes());
            }
        }

        //执行插入数据操作
        connectionTable.put(put);
        //释放资源
        connectionTable.close();
    }

    public static void deleteData(Connection connection, String nameSpace, String table, String rowKey) throws IOException {
        //获取表对象
        Table connectionTable = connection.getTable(TableName.valueOf(nameSpace + ":" + table));
        //创建Delete对象
        Delete delete = new Delete(rowKey.getBytes());
        //执行删除操作
        connectionTable.delete(delete);
        //释放资源
        connectionTable.close();
    }

    /**
     * @param rowKey     1001
     * @param sinkExtend "00|,01|,02|"
     * @return 00_1001  or  01_1001 ...
     */
    public static String getRowKey(String rowKey, String sinkExtend) {

        String[] split = sinkExtend.split(",");
        ArrayList<String> list = new ArrayList<>();
        for (String s : split) {
            list.add(s.replace("|", "_"));
        }
        list.add(split[split.length - 1]);

        return list.get(Math.abs(rowKey.hashCode() % list.size())) + rowKey;
    }

    public static String getBaseDicDDL() {
        return "CREATE TABLE dim_base_dic (\n" +
                "    rowkey string,\n" +
                "    info ROW<dic_name string>\n" +
                ") WITH (\n" +
                "    'connector' = 'hbase-2.2',\n" +
                "    'table-name' = 'gmall_230524:dim_base_dic',\n" +
                "    'zookeeper.quorum' = 'hadoop102:2181'\n" +
                ")";
    }

    public static JSONObject getData(Connection connection, String nameSpace, String tableName, String rowKey) throws IOException {
        //获取表对象
        Table table = connection.getTable(TableName.valueOf(nameSpace + ":" + tableName));
        //创建Get对象
        Get get = new Get(rowKey.getBytes());
        //执行Get操作
        Result result = table.get(get);
        //解析result,封装JSONObject
        JSONObject jsonObject = new JSONObject();
        Cell[] cells = result.rawCells();
        for (Cell cell : cells) {
            jsonObject.put(new String(CellUtil.cloneQualifier(cell)), new String(CellUtil.cloneValue(cell)));
        }
        //释放资源
        table.close();
        //返回数据
        return jsonObject;
    }

    public static JSONObject getData(AsyncConnection connection, String nameSpace, String tableName, String rowKey) throws IOException, ExecutionException, InterruptedException {
        //获取表对象
        AsyncTable<AdvancedScanResultConsumer> table = connection.getTable(TableName.valueOf(nameSpace + ":" + tableName));
        //创建Get对象
        Get get = new Get(rowKey.getBytes());
        //执行Get操作
        Result result = table.get(get).get();
        //解析result,封装JSONObject
        JSONObject jsonObject = new JSONObject();
        Cell[] cells = result.rawCells();
        for (Cell cell : cells) {
            jsonObject.put(new String(CellUtil.cloneQualifier(cell)), new String(CellUtil.cloneValue(cell)));
        }
        //返回数据
        return jsonObject;
    }

    public static void main(String[] args) throws Exception {
//        System.out.println(getRowKey("abc", "00|,01|,02|"));
//        System.out.println(getRowKey("def", "00|,01|,02|"));
//        System.out.println(getRowKey("saghsa", "00|,01|,02|"));
//        System.out.println(getRowKey("sajil", "00|,01|,02|"));

        Connection connection = getConnection();

        long start = System.currentTimeMillis();
        System.out.println(getData(connection, Constant.HBASE_NAME_SPACE, "dim_base_trademark", "10")); // 910 945 915 951
        long end1 = System.currentTimeMillis();
        System.out.println(getData(connection, Constant.HBASE_NAME_SPACE, "dim_base_trademark", "10")); //2 2 4   3*6=18
        long end2 = System.currentTimeMillis();

        System.out.println(end1 - start);
        System.out.println(end2 - end1);

        connection.close();
    }

}
